package mqttsp

import (
	"errors"
	"time"

	"gitee.com/yscsky/yu"
	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// IMqttSubPub MQTT订阅发布接口
type IMqttSubPub interface {
	SetSubDeal(func(mqtt.Message))
	Sub(topic string) (err error)
	Unsub(topic string) error
	Pub(topic string, payload []byte)
	SetOnConn(mqtt.OnConnectHandler)
}

type mqttSubPub struct {
	clientID string
	client   mqtt.Client
	subDeal  func(mqtt.Message)
	subQue   chan mqtt.Message
	pubQue   chan *pubArgs
	onConn   mqtt.OnConnectHandler
}

type pubArgs struct {
	topic   string
	payload []byte
}

// NewMqttSubPub 创建MqttSubPub
func NewMqttSubPub(id, addr, user, pass string) (imps IMqttSubPub, err error) {
	m := &mqttSubPub{
		clientID: id,
		subQue:   make(chan mqtt.Message, 1000),
		pubQue:   make(chan *pubArgs, 1000),
	}
	m.startQueue()
	opt := mqtt.NewClientOptions().
		AddBroker(addr).
		SetClientID(id).
		SetUsername(user).
		SetPassword(pass).
		SetCleanSession(true).
		SetAutoReconnect(true).
		SetKeepAlive(5 * time.Second).
		SetConnectTimeout(5 * time.Second).
		SetWriteTimeout(5 * time.Second).
		SetOnConnectHandler(m.onConnect).
		SetConnectionLostHandler(m.connectLost)
	m.client = mqtt.NewClient(opt)
	if err = m.connect(addr); err != nil {
		return
	}
	imps = m
	return
}

func (m *mqttSubPub) startQueue() {
	go func() {
		for msg := range m.subQue {
			if m.subDeal == nil {
				return
			}
			m.subDeal(msg)
		}
	}()
	go func() {
		for pa := range m.pubQue {
			// 检查mqtt连接是否正常
			if !m.client.IsConnected() {
				if token := m.client.Connect(); token.Wait() && token.Error() != nil {
					yu.Errf("client: %s, connect fail: %v", m.clientID, token.Error())
				}
			}
			if token := m.client.Publish(pa.topic, 2, false, pa.payload); token.Wait() && token.Error() != nil {
				yu.Errf("client: %s, publish fail: %v", m.clientID, token.Error())
			}
		}
	}()
}

func (m *mqttSubPub) connect(addr string) error {
	if m.client == nil {
		return errors.New("client is nil")
	}
	if m.client.IsConnected() {
		m.client.Disconnect(250)
	}
	if token := m.client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	yu.Logf("%s connect %s success", m.clientID, addr)
	return nil
}

func (m *mqttSubPub) onConnect(c mqtt.Client) {
	if m.onConn != nil {
		m.onConn(c)
	}
}

func (m *mqttSubPub) connectLost(c mqtt.Client, err error) {
	if err != nil {
		yu.Errf("client %s connect lost: %v", m.clientID, err)
	}
	yu.Logf("client %s try to reconnect", m.clientID)
}

// SetSubDeal 设置订阅处理回调
func (m *mqttSubPub) SetSubDeal(deal func(mqtt.Message)) {
	if deal != nil {
		m.subDeal = deal
	}
}

// Sub 订阅主题
func (m *mqttSubPub) Sub(topic string) (err error) {
	// 检查mqtt连接是否正常
	if !m.client.IsConnected() {
		if token := m.client.Connect(); token.Wait() && token.Error() != nil {
			return token.Error()
		}
	}
	if token := m.client.Subscribe(topic, 2, func(c mqtt.Client, msg mqtt.Message) {
		// 将订阅消息送入处理对列
		go func() { m.subQue <- msg }()
	}); token.Wait() && token.Error() != nil {
		err = token.Error()
	}
	return
}

// Pub 发布消息到MQTT
func (m *mqttSubPub) Pub(topic string, payload []byte) {
	// 放到队列中执行
	go func() { m.pubQue <- &pubArgs{topic: topic, payload: payload} }()
}

// Unsub 取消订阅主题
func (m *mqttSubPub) Unsub(topic string) error {
	// 检查mqtt连接是否正常
	if !m.client.IsConnected() {
		if token := m.client.Connect(); token.Wait() && token.Error() != nil {
			return token.Error()
		}
	}
	if token := m.client.Unsubscribe(topic); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// SetOnConn 设置连接时处理
func (m *mqttSubPub) SetOnConn(handle mqtt.OnConnectHandler) {
	if handle != nil {
		m.onConn = handle
	}
}
